package beautiful.butterfly.drds.schedule.zoo_keeper;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class Watcher implements org.apache.zookeeper.Watcher {
    private static transient Logger logger = LoggerFactory.getLogger(Watcher.class);
    private Map<String, org.apache.zookeeper.Watcher> pathToWatcherMap = new ConcurrentHashMap<String, org.apache.zookeeper.Watcher>();
    private ZooKeeperManager zooKeeperManager;

    public Watcher(ZooKeeperManager zooKeeperManager) {
        this.zooKeeperManager = zooKeeperManager;
    }

    public void registerChildrenChanged(String path, org.apache.zookeeper.Watcher watcher) throws Exception {
        zooKeeperManager.getZooKeeper().getChildren(path, true);
        pathToWatcherMap.put(path, watcher);
    }

    public void process(WatchedEvent watchedEvent) {
        if (logger.isInfoEnabled()) {
            logger.info("�Ѿ�������" + watchedEvent.getType() + ":" + watchedEvent.getState() + "�¼���" + watchedEvent.getPath());
        }
        if (watchedEvent.getType() == Event.EventType.NodeChildrenChanged) {
            String path = watchedEvent.getPath();
            org.apache.zookeeper.Watcher watcher = pathToWatcherMap.get(path);
            if (watcher != null) {
                try {
                    watcher.process(watchedEvent);
                } finally {
                    try {
                        if (zooKeeperManager.getZooKeeper().exists(path, null) != null) {
                            zooKeeperManager.getZooKeeper().getChildren(path, true);
                        }
                    } catch (Exception e) {
                        logger.error(path + ":" + e.getMessage(), e);
                    }
                }
            } else {
                logger.info("�Ѿ�������" + watchedEvent.getType() + ":" + watchedEvent.getState() + "�¼���" + watchedEvent.getPath());
            }
        } else if (watchedEvent.getState() == KeeperState.AuthFailed) {
            logger.info("tb_hj_schedule zoo_keeper status =KeeperState.AuthFailed��");
        } else if (watchedEvent.getState() == KeeperState.ConnectedReadOnly) {
            logger.info("tb_hj_schedule zoo_keeper status =KeeperState.ConnectedReadOnly��");
        } else if (watchedEvent.getState() == KeeperState.Disconnected) {
            logger.info("tb_hj_schedule zoo_keeper status =KeeperState.Disconnected��");
            try {
                zooKeeperManager.closeAndConnect();
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        } else if (watchedEvent.getState() == KeeperState.NoSyncConnected) {
            logger.info("tb_hj_schedule zoo_keeper status =KeeperState.NoSyncConnected���ȴ����½���ZK����.. ");
            try {
                zooKeeperManager.closeAndConnect();
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        } else if (watchedEvent.getState() == KeeperState.SaslAuthenticated) {
            logger.info("tb_hj_schedule zoo_keeper status =KeeperState.SaslAuthenticated��");
        } else if (watchedEvent.getState() == KeeperState.Unknown) {
            logger.info("tb_hj_schedule zoo_keeper status =KeeperState.Unknown��");
        } else if (watchedEvent.getState() == KeeperState.SyncConnected) {
            logger.info("�յ�ZK���ӳɹ��¼���");
        } else if (watchedEvent.getState() == KeeperState.Expired) {
            logger.error("�Ự��ʱ���ȴ����½���ZK����...");
            try {
                zooKeeperManager.closeAndConnect();
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        }
    }
}